package com.lagou.spark

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

import scala.collection.immutable

/*
  假设k=3
  算法：1、读文件 iris.csv
        2、随机找3个点作为初始中心点
        3、遍历数据集，计算每一个点与3个中心点的距离，距离哪个中心点近就属于哪个中心点
        4、根据新的分类计算新的中心点
        5、使用新的中心点开始下一次循环（回到第3步）
  退出循环条件：1、指定循环次数
               2、所有的中心点几乎不再移动，即中心点移动的最大距离小于给定常数（如0.01）
               修正：所有中心点移动距离的总和小于给定常数
 */

object KmeansDemo{
  def main(args: Array[String]) {

    val k = 3 //初始中心点个数k
    val minMove = 0.01 //minMove 中心点移动的最小距离，作为循环退出条件

    val conf = new SparkConf().setMaster("local[*]").setAppName("KMeans")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    // 读取文件
    val lines: RDD[String] = sc.textFile("data/Iris.csv")
    val data: RDD[LabelPoint] = lines.filter(_.trim.nonEmpty)
      .map(line => {
        val fields: Array[String] = line.split(",")
        LabelPoint(fields.last, fields.tail.init.map(_.toDouble)) //tail取集合中的除了第一个元素剩下的元素集合 init取集合中的除了最后一个元素剩下的元素集合
      })
    data.cache()
    //data.foreach(x=>println(x.label, x.point.toBuffer))

    // 找k个随机初始中心点
    val centerPoints: Array[Array[Double]] = data.takeSample(withReplacement = false, k).map(_.point)
    var tmpMove= 1.0

    while(tmpMove > minMove) {
      // 计算所有点到中心点的距离
      // 计算每个点的分类 [分类编号, (特征, 1.0)]
      val indexRDD: RDD[(Int, (Array[Double], Double))] = data
        .map(p => (getIndex(p.point, centerPoints), (p.point, 1.0)))

      // 所有点按分类计算出归属
      def arrayAdd(x: Array[Double], y: Array[Double]): Array[Double] = x.zip(y).map(elem => elem._1 + elem._2)
      val catalogRDD: RDD[(Int, (Array[Double], Double))] = indexRDD
        .reduceByKey((x, y) => (arrayAdd(x._1, y._1), x._2 + y._2)
      )

      // 计算新的中心点
      val newCenterPoints: collection.Map[Int, Array[Double]] = catalogRDD
        .map{
          case (index, (point, count)) => (index, point.map(_ / count))
        }.collectAsMap()

      // 计算中心点移动的距离之和
      val distance: immutable.Seq[Double] = for (i <- 0 until k) yield {
        getDistance(centerPoints(i), newCenterPoints(i))
      }
      tmpMove = distance.sum

      // 确定最终中心点
      for ((key, value) <- newCenterPoints) {
        centerPoints(key) = value
      }
      println(" distanceSum ：" + tmpMove + "")
    }

    // 输出结果
    println("最后确定的中心点:")
    centerPoints.foreach(x => println( x.toBuffer))

    sc.stop()
  }

  case class LabelPoint(label: String, point: Array[Double])

  // 距离
  def getDistance(x: Array[Double], y: Array[Double]): Double = {
    Math.sqrt(x.zip(y).map(elem => Math.pow(elem._1 - elem._2, 2)).sum)
  }

  // 分类
  def getIndex(p: Array[Double], centers: Array[Array[Double]]): Int = {
    val distance = centers.map(point => getDistance(point, p))
    distance.indexOf(distance.min)
  }
}